package com.winter.storm;

import com.winter.storm.bolt.myBolt;
import org.apache.storm.Config;
import org.apache.storm.StormSubmitter;
import org.apache.storm.kafka.*;
import org.apache.storm.spout.SchemeAsMultiScheme;
import org.apache.storm.topology.TopologyBuilder;

import java.util.UUID;

public class IndexTopology {




//    public static void main(String[] args) throws Exception {
//
//        TopologyBuilder builder = new TopologyBuilder();
//        builder.setSpout("spout", new SentenceSpout(), 1);
//        builder.setBolt("split", new SplitSentenceBolt(), 2).shuffleGrouping("spout");
//        builder.setBolt("count", new WordCountBolt(), 2).fieldsGrouping("split", new Fields("word"));
//
//        Config conf = new Config();
//        conf.setDebug(false);
//
//        if (args != null && args.length > 0) {
//            // 集群模式
//            conf.setNumWorkers(2);
//            StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
//        } else {
//            // 本地模式
//            LocalCluster cluster = new LocalCluster();
//            cluster.submitTopology("word-count", conf, builder.createTopology());
//            Thread.sleep(10000);
//            cluster.shutdown();
//        }
//    }




    public static void main(String[] args) {
        final TopologyBuilder topologybuilder = new TopologyBuilder();
        String topicName = "ac_interface";

        BrokerHosts hosts = new ZkHosts("");
        SpoutConfig spoutConfig = new SpoutConfig(hosts, topicName, "/" + topicName, UUID.randomUUID().toString());
        spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
        KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);


        // final TopologyBuilder topologybuilder = new TopologyBuilder();
        topologybuilder.setSpout("kafka_spout", kafkaSpout, 1);
        topologybuilder.setBolt("bolt", new myBolt()).shuffleGrouping("kafka_spout");

        Config stormConf=new Config();
        stormConf.setNumWorkers(1);
        stormConf.setDebug(true);
        try {
            System.setProperty("storm.jar", "storm-test.jar");
            StormSubmitter.submitTopology("testtopology", stormConf, topologybuilder.createTopology());
        } catch (Exception e) {
            // TODO Auto-generated catch block
            System.out.println("出现AuthorizationException");
            e.printStackTrace();
        }
    }


}
